package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author 杨晨昱
 * 课程主题 下单情况统计
 */
public class DwdCourseOrderDetail {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30));

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_course_order_detail_group"));
        //订单明细表
        Table orderDetail = tableEnv.sqlQuery("select " +
                "data['id'] id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['order_id'] order_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['final_amount']  final_amount,\n" +
                "data['session_id'] session_id,\n" +
                "ts\n" +
                " from `topic_db`\n " +
                " where `table` = 'order_detail'\n" +
                " and `type` = 'insert' \n");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        //订单表
        Table orderInfo = tableEnv.sqlQuery("select" +
                "  data['id'] id,\n" +
                "  data['user_id'] user_id,\n" +
                "  data['order_status'] order_status,\n" +
                "  data['province_id'] province_id\n" +
                " from `topic_db`\n " +
                " where `table` = 'order_info'\n" +
                " and `type` = 'insert'");
        tableEnv.createTemporaryView("order_info", orderInfo);

        Table tableResult = tableEnv.sqlQuery("select " +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.final_amount,\n" +
                "od.session_id,\n" +
                "oi.order_status,\n" +//订单状态
                "oi.province_id,\n" +
                "od.ts\n" +
                "from order_detail od \n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id\n"
        );
        tableEnv.createTemporaryView("table_result", tableResult);
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                " id String,   \n" +
                " course_id String,   \n" +
                " course_name String,   \n" +
                " order_id String,  \n" +
                " user_id String,  \n" +
                " final_amount String,   \n" +
                " session_id String,   \n" +
                        " order_status String,   \n" +
                        " province_id String , "+
                " ts String ,  \n"+
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from table_result");
        //course_info课程信息表
        //省份表
        //科目表
        //分类表
        tableEnv.executeSql("select * from dwd_trade_order_detail").print();
    }
}
